﻿////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
// <copyright>Copyright 2008-2011 Andreas Huber Doenni</copyright>
// Distributed under the Microsoft Public License (Ms-PL).
// See accompanying file License.txt or copy at http://mews.codeplex.com/license.
////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////

namespace Mews.Net.Nntp.Cache
{
    using System;
    using System.Collections;
    using System.Collections.Generic;
    using System.Data.SqlServerCe;
    using System.Diagnostics.CodeAnalysis;
    using System.IO;
    using System.Linq;
    using System.Net;
    using System.Threading;

    using Mews.IO;
    using Mews.Mime;

    /// <summary>Represents a local cache of information sent to or received from one or more NNTP servers.</summary>
    /// <threadsafety static="true" instance="false"/>
    [SuppressMessage("Microsoft.Maintainability", "CA1506:AvoidExcessiveClassCoupling", Justification = "There's no meaningful way to reduce the coupling.")]
    [SuppressMessage("Microsoft.Naming", "CA1724:TypeNamesShouldNotMatchNamespaces", Justification = "Applications also using the System.Net.Cache namespace can disambiguate by using Nntp.Cache.")]
    public sealed class Cache : IDisposable
    {
        private readonly ICryptoProvider cryptoProvider;
        private readonly int maxInboundArticleOverflow;
        private Database database;
        private readonly int minEncryptedLength;
        private Thread sendReceiveThread;

        private const string GroupIdName = "@groupid";
        private const string NewestOldNumberName = "@newestOldNumber";

        // This is to keep memory consumption at acceptable levels when receiving large groups
        private const long MaxBatchSize = 16384;
        private const int BufferSize = 4096;
        private static readonly byte[] lineBreak = new byte[] { (byte)'\r', (byte)'\n' };

        ////////////////////////////////////////////////////////////////////////////////////////////////////////////////

        /// <summary>Initializes a new instance of the <see cref="Cache"/> class by calling
        /// <see cref="Cache(string, ICryptoProvider, int)"/>(<paramref name="dataDirectory"/>,
        /// <paramref name="cryptoProvider"/>, 500).</summary>
        public Cache(string dataDirectory, ICryptoProvider cryptoProvider) : this(dataDirectory, cryptoProvider, 500)
        {
        }

        /// <summary>Initializes a new instance of the <see cref="Cache"/> class.</summary>
        /// <param name="dataDirectory">The directory where the cache data is located.</param>
        /// <param name="cryptoProvider">The provider that is used to encrypt &amp; decrypt the
        /// <see cref="IHost.AccountName"/> and <see cref="IHost.Password"/> properties in the cache.</param>
        /// <param name="maxInboundArticleOverflow">The total maximum number of excess articles that all overflowing
        /// groups can contain before <see cref="ShouldCompact"/> returns <c>true</c>.</param>
        /// <exception cref="ArgumentException"><paramref name="dataDirectory"/> contains at least one invalid
        /// character.</exception>
        /// <exception cref="ArgumentNullException"><paramref name="dataDirectory"/> and/or
        /// <paramref name="cryptoProvider"/> equal <c>null</c>.</exception>
        /// <exception cref="ArgumentOutOfRangeException"><paramref name="maxInboundArticleOverflow"/> is negative.
        /// </exception>
        /// <exception cref="FileNotFoundException"><paramref name="dataDirectory"/> does not exist or does not contain
        /// cache data.</exception>
        /// <exception cref="IncompatibleCacheException"><paramref name="dataDirectory"/> contains incompatible data.
        /// </exception>
        /// <remarks>
        /// <para>To create a new empty cache, call <see cref="CreateCache"/>.</para>
        /// </remarks>
        public Cache(string dataDirectory, ICryptoProvider cryptoProvider, int maxInboundArticleOverflow)
        {
            if (maxInboundArticleOverflow < 0)
            {
                throw new ArgumentOutOfRangeException("maxInboundArticleOverflow", "Must not be negative.");
            }

            this.cryptoProvider = cryptoProvider;
            this.maxInboundArticleOverflow = maxInboundArticleOverflow;
            this.database = new Database(this, dataDirectory, cryptoProvider);
            this.minEncryptedLength = this.database.Encrypt(string.Empty).Length;
        }

        /// <summary>Gets the directory passed to the constructor.</summary>
        public string DataDirectory
        {
            get { return this.database.DataDirectory; }
        }

        /// <summary>Gets the maximum length the value of the <see cref="IHost.Name"/> property can have.</summary>
        public int HostNameMaxLength
        {
            get { return this.database.HostUpdater.HostNameMaxLength; }
        }

        /// <summary>Gets the maximum length the value of the <see cref="IHost.UserName"/> property can have.</summary>
        public int UserNameMaxLength
        {
            get { return this.database.HostUpdater.UserNameMaxLength; }
        }

        /// <summary>Gets the maximum length the value of the <see cref="IHost.EmailAddress"/> property can have.
        /// </summary>
        public int EmailAddressMaxLength
        {
            get { return this.database.HostUpdater.EmailAddressMaxLength; }
        }

        /// <summary>Gets the maximum length the encrypted value of the <see cref="IHost.AccountName"/> property can
        /// have.</summary>
        public int AccountNameMaxLength
        {
            get { return this.database.HostUpdater.AccountNameMaxEncryptedLength - this.minEncryptedLength; }
        }

        /// <summary>Gets the maximum length the encrypted value of the <see cref="IHost.Password"/> property can have.
        /// </summary>
        public int PasswordMaxLength
        {
            get { return this.database.HostUpdater.PasswordMaxEncryptedLength - this.minEncryptedLength; }
        }

        /// <summary>Gets the maximum length the value of the <see cref="IGroup.Name"/> property can have.</summary>
        public int GroupNameMaxLength
        {
            get { return this.database.GroupUpdater.GroupNameMaxLength; }
        }

        /// <summary>Gets the maximum length the value of the <see cref="IGroup.DisplayName"/> property can have.
        /// </summary>
        public int GroupDisplayNameMaxLength
        {
            get { return this.database.GroupUpdater.GroupDisplayNameMaxLength; }
        }

        /// <summary>Gets the maximum length the value of the <see cref="IOutboundArticle.Subject"/> property can have.
        /// </summary>
        public int ArticleSubjectMaxLength
        {
            get { return this.database.OutboundArticleUpdater.SubjectMaxLength; }
        }

        /// <summary>Returns the return value of
        /// <see cref="AddHost(string, int, string, string, string, string)"/>(
        /// <paramref name="name"/>, <paramref name="port"/>, null, null, null, null).</summary>
        public IHost AddHost(string name, int port)
        {
            return this.AddHost(name, port, null, null, null, null);
        }

        /// <summary>Adds a new host to the cache and returns an object representing the host.</summary>
        /// <exception cref="ArgumentException">The length of either <paramref name="name"/>,
        /// <paramref name="userName"/> or <paramref name="emailAddress"/>, <paramref name="accountName"/> or
        /// <paramref name="password"/> exceeds the maximum allowed length, or the cache already contains a host with
        /// the same name and port.</exception>
        /// <exception cref="ArgumentNullException"><paramref name="name"/> equals <c>null</c>.</exception>
        /// <exception cref="ArgumentOutOfRangeException"><paramref name="port"/> is not between
        /// <see cref="IPEndPoint.MinPort"/> and <see cref="IPEndPoint.MaxPort"/>.</exception>
        /// <exception cref="ObjectDisposedException"><see cref="Dispose"/> has been called.</exception>
        public IHost AddHost(
            string name, int port, string userName, string emailAddress, string accountName, string password)
        {
            return this.database.AddHost(name, port, userName, emailAddress, accountName, password);
        }

        /// <summary>Gets the hosts that have been added with <see cref="AddHost(string, int)"/>.</summary>
        /// <exception cref="ObjectDisposedException"><see cref="Dispose"/> has been called.</exception>
        /// <remarks>The documented exceptions are never thrown from this method but from the
        /// <see cref="IEnumerator.MoveNext"/> method.</remarks>
        [SuppressMessage("Microsoft.Design", "CA1024:UsePropertiesWhereAppropriate", Justification = "This is an expensive operation.")]
        public IEnumerable<IHost> GetHosts()
        {
            foreach (var host in this.database.GetHosts())
            {
                yield return host;
            }
        }

        /// <summary>Gets the outbound articles that have the specified send status and where
        /// <see cref="IOutboundArticle.IsDeleted"/> equals <c>false</c>.</summary>
        /// <param name="sendStatus">The <see cref="IOutboundArticle.SendStatus"/> the returned objects must have.
        /// </param>
        /// <exception cref="ObjectDisposedException"><see cref="Dispose"/> has been called.</exception>
        /// <remarks>The documented exceptions are never thrown from this method but from the
        /// <see cref="IEnumerator.MoveNext"/> method.</remarks>
        [SuppressMessage("Microsoft.Design", "CA1024:UsePropertiesWhereAppropriate", Justification = "This is an expensive operation.")]
        public IEnumerable<IOutboundArticle> GetOutboundArticles(SendStatus sendStatus)
        {
            foreach (var article in this.database.GetOutboundArticles(sendStatus))
            {
                yield return article;
            }
        }

        /// <summary>Gets the outbound articles where <see cref="IOutboundArticle.IsDeleted"/> equals <c>true</c>.
        /// </summary>
        /// <exception cref="ObjectDisposedException"><see cref="Dispose"/> has been called.</exception>
        /// <remarks>The documented exceptions are never thrown from this method but from the
        /// <see cref="IEnumerator.MoveNext"/> method.</remarks>
        [SuppressMessage("Microsoft.Design", "CA1024:UsePropertiesWhereAppropriate", Justification = "This is an expensive operation.")]
        public IEnumerable<IOutboundArticle> GetDeletedOutboundArticles()
        {
            foreach (var article in this.database.GetDeletedOutboundArticles())
            {
                yield return article;
            }
        }

        /// <summary>Gets the number of articles where <see cref="IOutboundArticle.IsDeleted"/> equals <c>false</c> for
        /// each possible value of <see cref="SendStatus"/>.</summary>
        [SuppressMessage("Microsoft.Design", "CA1024:UsePropertiesWhereAppropriate", Justification = "This is an expensive operation.")]
        public IDictionary<SendStatus, int> GetOutboundArticleCounts()
        {
            return this.database.GetOutboundArticleCounts();
        }

        /// <summary>Starts a send/receive operation.</summary>
        /// <param name="groups">Articles from these groups will be received.</param>
        /// <param name="receiveFullArticles">Pass <c>true</c> to mark the newest articles such that the full article
        /// content will also be received in this send/receive operation.</param>
        /// <param name="observer">The observer to which send/receive progress will be reported.</param>
        /// <exception cref="ArgumentNullException"><paramref name="groups"/> and/or <paramref name="observer"/> equal
        /// <c>null</c>.</exception>
        /// <exception cref="InvalidOperationException">Another send/receive operation is already in progress.
        /// </exception>
        /// <remarks>
        /// <para><b>Caution</b>: The methods of the <paramref name="observer"/> object are called from a newly created
        /// thread. It is the duty of the <see cref="ISendReceiveProgressObserver"/> implementer to use proper
        /// mechanisms to avoid race conditions and/or deadlocks.</para>
        /// <para>Each send/receive operation has two phases:
        /// <list type="number">
        /// <item>Information about the newest articles is received for all specified groups. In any event,
        /// information of at most <see cref="IGroup.NewInboundArticlesReceiveLimit">IGroup.NewArticlesReceiveLimit</see>
        /// articles is received for each group. If articles from a given group have been received before, only the
        /// information about the articles posted since the last send/receive operation are retrieved. If
        /// <paramref name="receiveFullArticles"/> is <c>true</c> then each article is also marked for body receive in
        /// phase 2.</item>
        /// <item>The full content of all articles that have been marked for receive is retrieved. This includes the
        /// articles that have been marked for body receive in phase 1 as well as the articles that have been marked by
        /// setting <see cref="IInboundArticle.IsBeingWatched"/> to true.</item>
        /// </list></para>
        /// </remarks>
        public void StartSendReceive(
            IEnumerable<IGroup> groups, bool receiveFullArticles, ISendReceiveProgressObserver observer)
        {
            this.AssertSendReceiveNotInProgress();

            if (groups == null)
            {
                throw new ArgumentNullException("groups");
            }

            if (observer == null)
            {
                throw new ArgumentNullException("observer");
            }

            var groupIds = (from current in groups select ((Group)current).Id).ToList();
            this.sendReceiveThread = new Thread(
                () => DoSendReceive(this.DataDirectory, this.cryptoProvider, groupIds, receiveFullArticles, observer));
            this.sendReceiveThread.IsBackground = true;
            this.sendReceiveThread.Priority = ThreadPriority.BelowNormal;
            this.sendReceiveThread.Start();
        }

        /// <summary>Stops a previously started send/receive operation.</summary>
        public void StopSendReceive()
        {
            if (this.sendReceiveThread != null)
            {
                this.sendReceiveThread.Abort();
                this.sendReceiveThread.Join();
                this.sendReceiveThread = null;
            }
        }

        /// <summary>Gets a value indicating whether a send/receive operation is currently in progress.</summary>
        public bool IsSendReceiveInProgress
        {
            get
            {
                if (this.sendReceiveThread != null)
                {
                    if (!this.sendReceiveThread.Join(0))
                    {
                        return true;
                    }

                    this.sendReceiveThread = null;
                }

                return false;
            }
        }

        /// <summary>Gets a value indicating whether the cache should be compacted.</summary>
        public bool ShouldCompact
        {
            get
            {
                if (GetOverflowingGroups(this.database).Aggregate(
                    0L, (count, pair) => count + pair.Value - pair.Key.CachedInboundArticlesLimit) >
                    this.maxInboundArticleOverflow)
                {
                    return true;
                }

                using (var metaData = new MetaData(this.database))
                {
                    return this.ShouldCompactDatabase(metaData);
                }
            }
        }

        /// <summary>Compacts the cache.</summary>
        /// <returns><c>true</c> if the cache was compacted non-intrusively. All objects that have been directly or
        /// indirectly retrieved through this object remain valid (e.g. <see cref="IHost"/> objects). <c>false</c> if
        /// the cache was compacted intrusively, all child objects were rendered invalid.</returns>
        /// <exception cref="MissingComponentException"><see cref="Compact"/> was called on a device where the component
        /// sqlce.repl.ppc.wce5.armv4i.CAB is not installed. All child objects were rendered invalid.</exception>
        /// <exception cref="InsufficientDiskSpaceException">There is insufficient disk space to perform the compaction.
        /// All child objects were rendered invalid.</exception>
        /// <exception cref="InvalidOperationException">A send/receive operation is in progress. All child objects
        /// remain valid.</exception>
        /// <exception cref="IOException">At least one of the database files was locked by another process, see message
        /// for more information. All child objects were rendered invalid.</exception>
        public bool Compact()
        {
            this.AssertSendReceiveNotInProgress();

            using (var metaData = new MetaData(this.database))
            {
                metaData.InsertsAndDeletesSinceLastCompaction += DeleteOldInboundArticles(
                    this.database, from pair in GetOverflowingGroups(this.database) select pair.Key);

                if (!this.ShouldCompactDatabase(metaData))
                {
                    return true;
                }

                // We deliberately set the counter to 0 before compaction. This ensures that clients using the
                // ShouldCompact property will not immediately attempt another compaction if the following one fails
                // (e.g. because the replication component is not installed on a device)
                metaData.InsertsAndDeletesSinceLastCompaction = 0;
            }

            this.database.Dispose();

            using (var engine =
                new SqlCeEngine(Database.GetConnectionString(Database.GetDatabasePath(this.DataDirectory))))
            {
                try
                {
                    engine.Compact(null);
                }
                catch (SqlCeException ex)
                {
                    switch (ex.NativeError)
                    {
                        case 25035:
                            this.ReopenDatabase();
                            throw new IOException("The database is locked by another process.", ex);
                        case 25037:
                        case 25038:
                            // This is hard to simulate on the desktop, so these lines will never be covered
                            this.ReopenDatabase();
                            throw new InsufficientDiskSpaceException(Database.GetDatabasePath(this.DataDirectory), ex);
                        case 27777:
                            // This can only happen on the compact framework, so these lines will never be covered
                            this.ReopenDatabase();
                            throw new MissingComponentException("sqlce.repl.ppc.wce5.armv4i.CAB is not installed.", ex);
                        default:
                            throw;
                    }
                }
            }

            DeleteEmptyDirectories(this.DataDirectory);
            this.ReopenDatabase();
            return false;
        }

        /// <summary>See <see cref="IDisposable.Dispose"/>.</summary>
        /// <remarks>Calling this method immediately renders all objects invalid that have been directly or indirectly
        /// retrieved through this object (e.g. <see cref="IHost"/> and <see cref="IGroup"/> objects).</remarks>
        public void Dispose()
        {
            this.StopSendReceive();
            this.database.Dispose();
        }

        /// <summary>Creates an empty cache in <paramref name="dataDirectory"/>.</summary>
        /// <param name="dataDirectory">The directory where the data of the cache will be located. Must either be empty
        /// or must not yet exist.</param>
        /// <exception cref="ArgumentException"><paramref name="dataDirectory"/> is either not a legal directory path or
        /// is a non-empty directory.</exception>
        /// <exception cref="ArgumentNullException"><paramref name="dataDirectory"/> equals <c>null</c>.</exception>
        /// <exception cref="DirectoryNotFoundException">The specified path is invalid (for example, it is on an
        /// unmapped drive).</exception>
        /// <exception cref="IOException"><list type="bullet">
        /// <item>The directory specified by <paramref name="dataDirectory"/> is read-only, or</item>
        /// <item>The directory specified by <paramref name="dataDirectory"/> exceeds the maximum length. For example,
        /// on Windows-based platforms, paths must be less than 248 characters. Note that the cache will itself create a
        /// directory structure under <paramref name="dataDirectory"/>, so the length of
        /// <paramref name="dataDirectory"/> must be considerably lower than the maximum.</item>
        /// </list></exception>
        /// <exception cref="NotSupportedException"><paramref name="dataDirectory"/> contains a colon character (:) that
        /// is not part of a drive label ("C:\").</exception>
        /// <exception cref="UnauthorizedAccessException">The caller does not have the required permission.</exception>
        /// <remarks>If they do not yet exist, creates all directories and subdirectories as specified by
        /// <paramref name="dataDirectory"/>. Then creates a new empty cache in <paramref name="dataDirectory"/>, such
        /// that a <see cref="Cache"/> object can be created by passing <paramref name="dataDirectory"/>.</remarks>
        public static void CreateCache(string dataDirectory)
        {
            Database.CreateDatabase(dataDirectory);
        }

        ////////////////////////////////////////////////////////////////////////////////////////////////////////////////

        internal void AssertSendReceiveNotInProgress()
        {
            if (this.IsSendReceiveInProgress)
            {
                throw new InvalidOperationException("A send/receive operation is in progress.");
            }
        }

        ////////////////////////////////////////////////////////////////////////////////////////////////////////////////

        private bool ShouldCompactDatabase(MetaData metaData)
        {
            // * 10 means that there is roughly one database compaction for every 5 article deletion runs. This is
            // due to the fact that inserts and deletes count towards the number that is used as a metric for DB
            // fragmentation but deletion obviously only counts the articles that are deleted.
            return metaData.InsertsAndDeletesSinceLastCompaction > (long)this.maxInboundArticleOverflow * 10;
        }

        private void ReopenDatabase()
        {
            this.database = new Database(this, this.DataDirectory, this.cryptoProvider);
        }

        private static void DoSendReceive(
            string dataDirectory,
            ICryptoProvider cryptoProvider,
            IList<int> groupIds,
            bool receiveFullArticles,
            ISendReceiveProgressObserver observer)
        {
            string hostName = "Unknown";
            string groupName = hostName;

            try
            {
                byte[] buffer = new byte[BufferSize];

                using (var database = new Database(null, dataDirectory, cryptoProvider))
                {
                    Connection connection = new Connection();

                    try
                    {
                        foreach (var group in GetGroups(database, groupIds))
                        {
                            hostName = group.Host.Name;
                            groupName = group.Name;

                            try
                            {
                                connection = ConnectClient(group.Host, connection, observer);
                            }
                            catch (IOException ex)
                            {
                                throw new HostUnreachableException("Host is unreachable.", ex);
                            }

                            if (connection.Host != null)
                            {
                                try
                                {
                                    SendArticles(connection.Client, group.Host, observer);
                                    ReceiveHeaders(connection.Client, group, receiveFullArticles, buffer, observer);
                                    ReceiveBodies(connection.Client, group, buffer, observer);
                                }
                                catch (WriteFailedException)
                                {
                                    throw;
                                }
                                catch (IOException ex)
                                {
                                    throw new ConnectionLostException("Connection has been lost.", ex);
                                }
                            }
                        }
                    }
                    finally
                    {
                        connection.Dispose();
                    }
                }
            }
            catch (ThreadAbortException)
            {
                observer.OnStopped();
            }
            catch (Exception ex)
            {
                if (!observer.OnError(hostName, groupName, ex))
                {
                    // Trying to cover this line reproducibly crashes the VS test host in release mode
                    throw;
                }
            }

            observer.OnFinished();
        }

        private static int DeleteOldInboundArticles(Database database, IEnumerable<Group> overflowingGroups)
        {
            const string RemoveInboundArticleReferencesCommandText =
                "UPDATE " + InboundArticleNames.Table + " SET " + InboundArticleNames.ReferencesNumberColumn +
                " = NULL WHERE " + InboundArticleNames.GroupIdColumn + " = " + GroupIdName + " AND " +
                InboundArticleNames.ReferencesNumberColumn + " <= " + NewestOldNumberName;
            const string RemoveGroupReferencesCommandText =
                "UPDATE [" + GroupNames.Table + "] SET " + GroupNames.SelectedInboundArticleNumberColumn + " = NULL WHERE " +
                GroupNames.IdColumn + " = " + GroupIdName + " AND " + GroupNames.SelectedInboundArticleNumberColumn +
                " <= " + NewestOldNumberName;

            int deletedCount = 0;

            using (var newestCommand = CommandHelper.Create(database.Connection, InboundArticleNames.Table))
            using (var removeInboundArticleReferencesCommand =
                new SqlCeCommand(RemoveInboundArticleReferencesCommandText, database.Connection))
            using (var removeGroupReferencesCommand =
                new SqlCeCommand(RemoveGroupReferencesCommandText, database.Connection))
            using (var removeInboundArticlesCommand =
                CommandHelper.Create(database.Connection, InboundArticleNames.Table))
            {
                AddRemoveParameters(removeInboundArticleReferencesCommand);
                AddRemoveParameters(removeGroupReferencesCommand);

                foreach (var group in overflowingGroups)
                {
                    var newestNumber = GetNewestNumber(newestCommand, group);

                    if (newestNumber != 0)
                    {
                        var newestOldNumber = newestNumber - group.CachedInboundArticlesLimit;
                        ExecuteRemoveCommand(removeInboundArticleReferencesCommand, group, newestOldNumber);
                        ExecuteRemoveCommand(removeGroupReferencesCommand, group, newestOldNumber);

                        removeInboundArticlesCommand.SetRange(
                            DbRangeOptions.Default,
                            new object[] { group.Id, 1 },
                            new object[] { group.Id, newestOldNumber });

                        using (var removeResultSet = removeInboundArticlesCommand.ExecuteResultSet(
                            ResultSetOptions.Scrollable | ResultSetOptions.Updatable))
                        {
                            while (removeResultSet.ReadFirst())
                            {
                                group.DeleteInboundArticleFile((long)removeResultSet[InboundArticleNames.NumberColumn]);
                                removeResultSet.Delete();
                                ++deletedCount;
                            }
                        }
                    }
                }
            }

            return deletedCount;
        }

        private static IList<Group> GetGroups(Database database, IList<int> groupIds)
        {
            List<Group> groups = new List<Group>(groupIds.Count);

            using (var hostCommand = CommandHelper.Create(database.Connection, HostNames.Table))
            using (var groupCommand = CommandHelper.Create(database.Connection, GroupNames.Table))
            using (var hostReader = hostCommand.ExecuteReader())
            using (var groupReader = groupCommand.ExecuteReader())
            {
                Host host = null;

                foreach (var groupId in groupIds)
                {
                    if (!groupReader.SeekAndRead(DbSeekOptions.FirstEqual, groupId))
                    {
                        throw new ArgumentException("The group with the Id " + groupId + " does not exist.");
                    }

                    int hostId = (int)groupReader[GroupNames.HostIdColumn];

                    if ((host == null) || (host.Id != hostId))
                    {
                        hostReader.SeekAndRead(DbSeekOptions.FirstEqual, hostId);
                        host = new Host(database, hostReader);
                    }

                    groups.Add(new Group(host, groupReader));
                }
            }

            return groups;
        }

        private static Connection ConnectClient(
            Host nextHost, Connection currentConnection, ISendReceiveProgressObserver observer)
        {
            if (nextHost.Equals(currentConnection.Host))
            {
                return currentConnection;
            }

            currentConnection.Dispose();

            using (var connectionGuard = ScopeGuard.Create(observer.OnConnecting(nextHost.Name)))
            {
                if (connectionGuard.Resource == null)
                {
                    return new Connection();
                }

                using (var clientGuard = ScopeGuard.Create(new Client(nextHost.Name, nextHost.Port)))
                using (var resultGuard =
                    ScopeGuard.Create(new Connection(nextHost, clientGuard.Resource, connectionGuard.Resource)))
                {
                    if (!string.IsNullOrEmpty(nextHost.AccountName))
                    {
                        new AuthInfoCommand(nextHost.AccountName, nextHost.Password).Execute(clientGuard.Resource);
                    }

                    observer.OnConnected(nextHost.Name);
                    connectionGuard.Dismiss();
                    clientGuard.Dismiss();
                    resultGuard.Dismiss();
                    return resultGuard.Resource;
                }
            }
        }

        private static void SendArticles(Client client, Host host, ISendReceiveProgressObserver observer)
        {
            var articles = host.GetOutboundArticles(SendStatus.Outbox).ToList();

            if (articles.Count > 0)
            {
                bool failureReported = false;
                int previousPercentage = 0;
                observer.OnSendProgressChanged(0);

                for (var index = 0; index < articles.Count; ++index)
                {
                    var message = articles[index].LoadMessage();
                    articles[index].SaveMessage(message); // Ensure that the date of the message is the current date

                    try
                    {
                        new PostCommand(stream => message.WriteTo(stream)).Execute(client);
                        host.Database.OutboundArticleUpdater.Update(articles[index].Id, SendStatus.Sent);
                    }
                    catch (ClientException ex)
                    {
                        if (!failureReported)
                        {
                            failureReported = true;

                            if (!observer.OnError(
                                host.Name, ((UnstructuredHeaderField)message.Fields["Newsgroups"].First()).Body, ex))
                            {
                                throw;
                            }
                        }
                    }

                    int currentPercentage = (index + 1) * 100 / articles.Count;

                    if (currentPercentage > previousPercentage)
                    {
                        previousPercentage = currentPercentage;
                        observer.OnSendProgressChanged(currentPercentage);
                    }
                }
            }
        }

        private static void ReceiveHeaders(
            Client client, Group group, bool receiveFullArticles, byte[] buffer, ISendReceiveProgressObserver observer)
        {
            var groupResponse = new GroupCommand(group.Name).Execute(client);
            var lastOldArticleNumber = NavigateToFirstNewArticle(
                client, groupResponse, group, GetNewestArticleNumber(group)) - 1;

            var totalArticleCount = groupResponse.HighWaterMark - lastOldArticleNumber;

            if (totalArticleCount <= 0)
            {
                return;
            }

            observer.OnHeadersProgressChanged(group.DisplayName, 0);
            int insertCount = 0;

            using (var inserter = new InboundArticleInserter(group.Connection))
            {
                bool lastArticleRead = false;
                var previousArticleNumber = lastOldArticleNumber;
                int previousPercentage = 0;
                string currentDirectory = null;

                while (!lastArticleRead && (previousArticleNumber < groupResponse.HighWaterMark))
                {
                    EnqueueHeaderCommandBatch(client, previousArticleNumber, groupResponse.HighWaterMark);

                    while (client.PendingCommandsCount > 0)
                    {
                        IResponse response;

                        try
                        {
                            response = client.GetResponse();
                        }
                        catch (NoNextArticleException)
                        {
                            lastArticleRead = true;
                            continue;
                        }

                        using (var headResponse = response as HeadResponse)
                        {
                            if ((headResponse == null) || lastArticleRead)
                            {
                                continue;
                            }

                            var path = InboundArticle.GetDataFilePath(group.DataDirectory, headResponse.Number);
                            var directory = Path.GetDirectoryName(path);

                            if (directory != currentDirectory)
                            {
                                Directory.CreateDirectory(directory);
                                currentDirectory = directory;
                            }

                            Message message = SaveAndParseHeader(headResponse.Content, path, buffer);

                            try
                            {
                                inserter.Insert(
                                    group,
                                    headResponse.Number,
                                    headResponse.MessageId,
                                    GetDirectReference(message),
                                    message.Date,
                                    message.FromField == null ? null : message.FromField.GetBody(),
                                    message.Subject,
                                    true,
                                    receiveFullArticles,
                                    null);
                                ++insertCount;
                            }
                            catch (ArgumentException)
                            {
                                // This can only happen as a result of the server sending two distinct messages with the
                                // same message-ID or the same number, which is clearly illegal. We therefore do what
                                // the server should have done at his end.
                                group.DeleteInboundArticleFile(headResponse.Number);
                            }

                            int currentPercentage =
                                (int)((headResponse.Number - lastOldArticleNumber) * 100 / totalArticleCount);

                            if (currentPercentage > previousPercentage)
                            {
                                previousPercentage = currentPercentage;
                                observer.OnHeadersProgressChanged(group.DisplayName, currentPercentage);
                            }

                            previousArticleNumber = headResponse.Number;
                        }
                    }
                }
            }

            if (insertCount > 0)
            {
                using (var metaData = new MetaData(group.Host.Database))
                {
                    metaData.InsertsAndDeletesSinceLastCompaction += insertCount;
                }
            }
        }

        private static void ReceiveBodies(
            Client client, Group group, byte[] buffer, ISendReceiveProgressObserver observer)
        {
            int totalArticleCount = GetPendingReceiveCount(group);

            if (totalArticleCount == 0)
            {
                return;
            }

            observer.OnBodiesProgressChanged(group.DisplayName, 0);
            int previousPercentage = 0;

            using (var articleCommand = CommandHelper.Create(
                group.Connection,
                InboundArticleNames.Table,
                InboundArticleNames.GroupIdReceiveFullArticleIndex,
                DbRangeOptions.Match,
                new object[] { group.Id, true },
                null))
            {
                while (EnqueueBodyCommandsBatch(client, articleCommand))
                {
                    while (client.PendingCommandsCount > 0)
                    {
                        using (var articleResultSet =
                            articleCommand.ExecuteResultSet(ResultSetOptions.Updatable | ResultSetOptions.Scrollable))
                        {
                            articleResultSet.ReadFirst();
                            var expectedArticleNumber = (long)articleResultSet[InboundArticleNames.NumberColumn];

                            try
                            {
                                using (var response = (BodyResponse)client.GetResponse())
                                {
                                    if (expectedArticleNumber != response.Number)
                                    {
                                        throw new UnexpectedResponseException(
                                            "The server responded with an unexpected article.");
                                    }

                                    SaveBody(group, response, buffer);
                                }

                                articleResultSet.SetValue(InboundArticleNames.HasBodyBeenReceivedColumn, true);
                                articleResultSet.SetValue(InboundArticleNames.ReceiveFullArticleColumn, false);
                                articleResultSet.Update();
                            }
                            catch (NoSuchArticleException)
                            {
                                // This can happen when an article is removed between overview/header & body
                                // receive
                                group.DeleteInboundArticleFile(expectedArticleNumber);
                                UpdateReferencingArticles(group, expectedArticleNumber);

                                // This also deletes the associated rows in the MessageId table
                                articleResultSet.Delete();
                            }
                        }

                        int currentPercentage =
                            (totalArticleCount - client.PendingCommandsCount) * 100 / totalArticleCount;

                        if (currentPercentage > previousPercentage)
                        {
                            previousPercentage = currentPercentage;
                            observer.OnBodiesProgressChanged(group.DisplayName, currentPercentage);
                        }
                    }
                }
            }
        }

        private static void AddRemoveParameters(SqlCeCommand removeReferencesCommand)
        {
            removeReferencesCommand.Parameters.Add(GroupIdName, 0);
            removeReferencesCommand.Parameters.Add(NewestOldNumberName, 0);
        }

        private static IEnumerable<KeyValuePair<Group, long>> GetOverflowingGroups(Database database)
        {
            const string ArticleCountName = "ArticleCount";
            const string CommandText =
                "SELECT [" + GroupNames.Table + "].*, " + ArticleCountName + "\r\nFROM [" + GroupNames.Table + "] INNER JOIN " +
                "(SELECT " + InboundArticleNames.GroupIdColumn + ", MAX(" + InboundArticleNames.NumberColumn +
                ") - MIN(" + InboundArticleNames.NumberColumn + ") + 1 AS " + ArticleCountName +
                " FROM " + InboundArticleNames.Table + " GROUP BY " + InboundArticleNames.GroupIdColumn +
                ") AS Counts ON " + GroupNames.IdColumn + " = " + InboundArticleNames.GroupIdColumn + "\r\n" +
                "WHERE NOT EXISTS (SELECT " + InboundArticleNames.NumberColumn + " FROM " + InboundArticleNames.Table +
                " WHERE " + InboundArticleNames.GroupIdColumn + " = [" + GroupNames.Table + "]." + GroupNames.IdColumn +
                " AND " + InboundArticleNames.HasBeenReadColumn + " = CONVERT(BIT, 0)) AND " + ArticleCountName + " - " +
                GroupNames.CachedInboundArticlesLimitColumn + " > 0";

            using (var groupCommand = new SqlCeCommand(CommandText, database.Connection))
            using (var groupReader = groupCommand.ExecuteReader())
            using (var hostCommand = CommandHelper.Create(database.Connection, HostNames.Table))
            using (var hostReader = hostCommand.ExecuteReader())
            {
                while (groupReader.Read() &&
                    hostReader.SeekAndRead(DbSeekOptions.FirstEqual, groupReader[GroupNames.HostIdColumn]))
                {
                    yield return new KeyValuePair<Group, long>(
                        new Group(new Host(database, hostReader), groupReader), (long)groupReader[ArticleCountName]);
                }
            }
        }

        private static long GetNewestNumber(SqlCeCommand maxCommand, Group group)
        {
            maxCommand.SetRange(DbRangeOptions.Prefix, new object[] { group.Id }, null);

            using (var maxResultSet = maxCommand.ExecuteResultSet(ResultSetOptions.Scrollable))
            {
                return maxResultSet.ReadLast() ? (long)maxResultSet[InboundArticleNames.NumberColumn] : 0;
            }
        }

        private static void ExecuteRemoveCommand(SqlCeCommand removeReferencesCommand, Group group, long newestOldNumber)
        {
            removeReferencesCommand.Parameters[GroupIdName].Value = group.Id;
            removeReferencesCommand.Parameters[NewestOldNumberName].Value = newestOldNumber;
            removeReferencesCommand.ExecuteNonQuery();
        }

        private static long NavigateToFirstNewArticle(
            Client client, GroupResponse groupResponse, Group group, long lastReceivedArticleNumber)
        {
            // Servers might report an empty group in different ways. One way is when the article count is 0 while the
            // low and high water marks might suggest the existence of articles, see
            // http://www.ietf.org/rfc/rfc3977.txt, 6.1.1.2.
            if (groupResponse.ArticleCount == 0)
            {
                return groupResponse.HighWaterMark + 1;
            }

            long firstNewArticleNumberCandidate = Math.Max(
                groupResponse.LowWaterMark,
                Math.Max(groupResponse.HighWaterMark - group.NewInboundArticlesReceiveLimit, lastReceivedArticleNumber) + 1);

            for (; firstNewArticleNumberCandidate <= groupResponse.HighWaterMark; ++firstNewArticleNumberCandidate)
            {
                try
                {
                    return new StatCommand(firstNewArticleNumberCandidate).Execute(client).Number;
                }
                catch (NoSuchArticleException)
                {
                }
            }

            return firstNewArticleNumberCandidate;
        }

        private static long GetNewestArticleNumber(Group group)
        {
            using (var newestCommand = CommandHelper.Create(group.Connection, InboundArticleNames.Table))
            {
                return GetNewestNumber(newestCommand, group);
            }
        }

        private static void EnqueueHeaderCommandBatch(Client client, long previousArticleNumber, long highWaterMark)
        {
            // In integer arithmetic 1 / 2 == 0, so we need to + 1 to ensure the result is always positive
            var batchSize = Math.Min((highWaterMark - previousArticleNumber + 1) / 2, MaxBatchSize / 2);

            for (int currentCount = 1; currentCount <= batchSize; ++currentCount)
            {
                client.EnqueueCommand(new HeadCommand());

                if (previousArticleNumber + currentCount < highWaterMark)
                {
                    client.EnqueueCommand(new NextCommand());
                }
            }
        }

        private static Message SaveAndParseHeader(Stream content, string path, byte[] buffer)
        {
            return WrapFileIOException(
                () =>
                {
                    using (var stream = File.Create(path, BufferSize))
                    {
                        int read;

                        while ((read = WrapNetworkIOException(() => content.Read(buffer, 0, buffer.Length))) > 0)
                        {
                            stream.Write(buffer, 0, read);
                        }

                        stream.Seek(0, SeekOrigin.Begin);
                        return Message.Parse(stream, BufferSize);
                    }
                },
                path);
        }

        private static T WrapFileIOException<T>(Func<T> func, string message)
        {
            try
            {
                return func();
            }
            catch (ConnectionLostException)
            {
                throw;
            }
            catch (IOException ex)
            {
                // Apparently, it is not possible to get a more detailed cause for the IOException, not even by
                // calling Marshal.GetHRForException. We therefore can't do anything better than report that
                // storage failed.
                throw new WriteFailedException(message, ex);
            }
        }

        private static void WrapFileIOException(Action action, string message)
        {
            WrapFileIOException(() => { action(); return false; }, message);
        }

        private static T WrapNetworkIOException<T>(Func<T> func)
        {
            try
            {
                return func();
            }
            catch (IOException ex)
            {
                throw new ConnectionLostException("Connection has been lost.", ex);
            }
        }

        private static string GetDirectReference(Message header)
        {
            // See http://www.ietf.org/rfc/rfc5322.txt, 3.6.4
            var references = header.References;

            if ((references != null) && (references.Count > 0))
            {
                return references[references.Count - 1];
            }

            var inReplyTo = header.InReplyTo;

            if ((inReplyTo != null) && (inReplyTo.Count == 1))
            {
                return inReplyTo[0];
            }

            return null;
        }

        private static int GetPendingReceiveCount(Group group)
        {
            var commandText = "SELECT COUNT(*) FROM " + InboundArticleNames.Table + " WITH(INDEX(" +
                InboundArticleNames.GroupIdReceiveFullArticleIndex + ")) WHERE " + InboundArticleNames.GroupIdColumn + " = " +
                GroupIdName + " AND " + InboundArticleNames.ReceiveFullArticleColumn + " = 1";

            using (var command = new SqlCeCommand(commandText, group.Connection))
            {
                command.Parameters.Add(GroupIdName, group.Id);
                return (int)command.ExecuteScalar();
            }
        }

        private static bool EnqueueBodyCommandsBatch(Client client, SqlCeCommand articleCommand)
        {
            using (var articleReader = articleCommand.ExecuteReader())
            {
                for (int count = 0; (count < MaxBatchSize) && articleReader.Read(); ++count)
                {
                    client.EnqueueCommand(new BodyCommand((long)articleReader[InboundArticleNames.NumberColumn]));
                }
            }

            return client.PendingCommandsCount > 0;
        }

        private static void SaveBody(Group group, BodyResponse response, byte[] buffer)
        {
            string path = InboundArticle.GetDataFilePath(group.DataDirectory, response.Number);

            WrapFileIOException(
                () =>
                {
                    using (var stream = new FileStream(path, FileMode.Append))
                    {
                        stream.Write(lineBreak, 0, lineBreak.Length);
                        int read;

                        while ((read =
                            WrapNetworkIOException(() => response.Content.Read(buffer, 0, buffer.Length))) > 0)
                        {
                            stream.Write(buffer, 0, read);
                        }
                    }
                },
                path);
        }

        private static void UpdateReferencingArticles(Group group, long articleNumber)
        {
            using (var command = CommandHelper.Create(
                group.Connection,
                InboundArticleNames.Table,
                InboundArticleNames.GroupIdReferencesNumberIndex,
                DbRangeOptions.Match,
                new object[] { group.Id, articleNumber },
                null))
            using (var resultSet = command.ExecuteResultSet(ResultSetOptions.Updatable))
            {
                while (resultSet.Read())
                {
                    resultSet.SetValue(InboundArticleNames.ReferencesNumberColumn, null);
                    resultSet.Update();
                }
            }
        }

        private static void DeleteEmptyDirectories(string directory)
        {
            foreach (var subDirectory in Directory.GetDirectories(directory))
            {
                DeleteEmptyDirectories(subDirectory);
            }

            if ((Directory.GetFiles(directory).Length == 0) && (Directory.GetDirectories(directory).Length == 0))
            {
                Directory.Delete(directory);
            }
        }

        ////////////////////////////////////////////////////////////////////////////////////////////////////////////////

        /// <summary>Bundles the connections necessary to talk to a NNTP host.</summary>
        private struct Connection : IDisposable
        {
            private Host host;
            private Client client;
            private readonly IDisposable networkConnection;

            ////////////////////////////////////////////////////////////////////////////////////////////////////////////

            /// <summary>See <see cref="IDisposable.Dispose"/>.</summary>
            public void Dispose()
            {
                if (this.host != null)
                {
                    this.host = null;
                    this.client.Dispose();
                    this.client = null;
                    this.networkConnection.Dispose();
                }
            }

            ////////////////////////////////////////////////////////////////////////////////////////////////////////////

            internal Connection(Host host, Client client, IDisposable networkConnection)
            {
                this.host = host;
                this.client = client;
                this.networkConnection = networkConnection;
            }

            internal Host Host
            {
                get { return this.host; }
            }

            internal Client Client
            {
                get { return this.client; }
            }
        }
    }
}
